package org.example.common;

import io.grpc.stub.StreamObserver;
import org.example.sec11.Sec02ExecutorCallOptionsTest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ResponseObserver<T> implements StreamObserver<T> {
    private static final Logger LOG = LoggerFactory.getLogger(ResponseObserver.class);

    private final List<T> list = Collections.synchronizedList(new ArrayList<>());
    private Throwable throwable;
    private final CountDownLatch latch;

    public ResponseObserver(int latch) {
        this.latch = new CountDownLatch(latch);
    }

    public static <T> ResponseObserver<T> create() {
        return new ResponseObserver<>(1);
    }

    public static <T> ResponseObserver<T> create(int latch) {
        return new ResponseObserver<>(latch);
    }

    @Override
    public void onNext(T t) {
        LOG.info("============>>>>>>>>received is:{}" , t);
        this.list.add(t);
    }

    @Override
    public void onError(Throwable throwable) {
        this.throwable = throwable;
        this.latch.countDown();
    }

    @Override
    public void onCompleted() {
        this.latch.countDown();
    }

    public  void await(){
        try {
            this.latch.await(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    public List<T> getItems(){
        return this.list;
    }

    public Throwable getThrowable() {
        return throwable;
    }



}
